/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once

#include "execution_defs.h"
#include "common/common.h"
#include "index/ix.h"
#include "system/sm.h"

class AbstractExecutor {
public:
    Rid _abstract_rid;

    Context* context_;

    virtual ~AbstractExecutor() = default;

    virtual size_t tupleLen() const { return 0; };

    virtual const std::vector<ColMeta>& cols() const {
        std::vector<ColMeta>* _cols = nullptr;
        return *_cols;
    };

    virtual std::string getType() { return "AbstractExecutor"; };

    virtual void beginTuple() {};

    virtual void nextTuple() {};

    virtual bool is_end() const { return true; }

    virtual Rid& rid() = 0;

    virtual std::unique_ptr<RmRecord> Next() = 0;

    virtual ColMeta get_col_offset(const TabCol& target) { return ColMeta(); };

    //检查记录是否满足单个条件 给scan与proj用
    bool test_single_cond(RmRecord& record, Condition& cond)
    {
        //colmeta
        auto& lcolmeta = *get_col(cols(), cond.lhs_col);

        char* lval = record.data + lcolmeta.offset;
        char* rval;
        ColType type;

        if (cond.is_rhs_val)
        {
            //若为值 从std::shared_ptr<RmRecord>强转
            type = cond.rhs_val.type;
            rval = (char*)cond.rhs_val.raw->data;
        }
        else
        {
            //若为列 则找到对应数据指针与偏移
            auto& rcolmeta = *get_col(cols(), cond.lhs_col);
            rval = record.data + rcolmeta.offset;
            type = rcolmeta.type;
        }

        assert(type == lcolmeta.type && "executor_abstract: condtion type unequal");

        int cmp;
        switch (type)
        {
            //取消用减法 防止溢出
        case TYPE_INT:
        {
            int il = *(int*)lval;
            int ir = *(int*)rval;
            // cmp=il-ir;
            cmp = (il < ir) ? -1 : ((il > ir) ? 1 : 0);
            break;
        }
        case TYPE_BIGINT:
        {
            long long bl = *(long long*)lval;
            long long br = *(long long*)rval;
            cmp = (bl < br) ? -1 : ((bl > br) ? 1 : 0);
            break;
        }
        case TYPE_FLOAT:
        {
            float fl = *(float*)lval;
            float fr = *(float*)rval;
            cmp = (fl < fr) ? -1 : ((fl > fr) ? 1 : 0);
            break;
        }
        case TYPE_STRING:
        {
            // std::string sl=*(std::string*)lval;
            // std::string sr=*(std::string*)rval;
            // cmp=sl.compare(sr);
            cmp = memcmp(lval, rval, lcolmeta.len);
            break;
        }
        case TYPE_TIME:
        {
            cmp = memcmp(lval, rval, lcolmeta.len);
        }
        }

        bool res = false;
        switch (cond.op)
        {
        case OP_EQ:
            res = cmp == 0;
            break;
        case OP_NE:
            res = cmp != 0;
            break;
        case OP_LT:
            res = cmp < 0;
            break;
        case OP_GT:
            res = cmp > 0;
            break;
        case OP_LE:
            res = cmp <= 0;
            break;
        case OP_GE:
            res = cmp >= 0;
            break;
        }
        return res;
    }

    //检查记录是否满足所有条件
    bool test_all_conds(RmRecord& record, std::vector<Condition>& conds)
    {
        for (auto cond : conds)
        {
            //一旦有一个条件不满足 跳出返回假
            if (!test_single_cond(record, cond))
            {
                return false;
            }
        }
        return true;
    }


    std::vector<ColMeta>::const_iterator get_col(const std::vector<ColMeta>& rec_cols, const TabCol& target) {
        auto pos = std::find_if(rec_cols.begin(), rec_cols.end(), [&](const ColMeta& col) {
            return col.tab_name == target.tab_name && col.name == target.col_name;
            });
        if (pos == rec_cols.end()) {
            throw ColumnNotFoundError(target.tab_name + '.' + target.col_name);
        }
        return pos;
    }
};